package com.liangliang.mock;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;
import java.util.Timer;
import java.util.TimerTask;

public class KafkaDataProducer {

    private static final String BROKER_LIST = "niit-master:9092";
    private static final String TOPIC = "shixuntopic";
    private static final String[] PROVINCES_CITIES = {
            "北京,北京",
            "上海,上海",
            "南京,江苏省",
            "广州,广东省",
            "三亚,海南省",
            "武汉,湖北省",
            "长沙,湖南省",
            "西安,陕西省",
            "成都,四川省",
            "哈尔滨,东北省"
    };

    public void startProducing() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        Timer timer = new Timer();
        timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                long timestamp = System.currentTimeMillis();
                int userId = new Random().nextInt(100);
                int adId = new Random().nextInt(20);
                String[] provinceCity = PROVINCES_CITIES[new Random().nextInt(PROVINCES_CITIES.length)].split(",");
                String province = provinceCity[0];
                String city = provinceCity[1];

                String message = timestamp + "," + province + "," + city + "," + userId + "," + adId;
                producer.send(new ProducerRecord<>(TOPIC, message));
                System.out.println("Sent message: " + message);
            }
        }, 0, 5000);

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            producer.close();
            timer.cancel();
        }));
    }
}
